package com.shujia.spark.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02WordCntOnYarn {
  def main(args: Array[String]): Unit = {
    /**
     * 提交到集群的一般步骤：
     * 1、注释conf.setMaster("local")
     * 2、将本地相对路径换成HDFS的路径，并准备数据
     * 3、打包上传程序的jar包
     * 4、使用命令进行提交：
     * spark-submit --master yarn --deploy-mode client --class com.shujia.spark.core.Demo02WordCntOnYarn spark-1.0.jar /data/spark/input01 /data/spark/output01
     */

    // 从args中取得提交命令中传入的参数
    if (args.length != 2) {
      println("请检查参数的个数，必须包含输入路径、输出路径")
      return
    }

    val inputPath: String = args(0)
    val outputPath: String = args(1)

    // 1、创建Spark环境
    val conf: SparkConf = new SparkConf() // Spark配置对象
    conf.setAppName("Demo02WordCntOnYarn")
    // local[8] ： 8个线程
    //    conf.setMaster("local") // 设置Spark的运行方式，如果不设置默认以集群方式运行

    val sc: SparkContext = new SparkContext(conf) // Spark程序的入口

    // 2、读文件
    /**
     * RDD：弹性分布式数据集（分布式的数据容器）
     * RDD它是一种编程模型，它并不会存储数据
     * 在使用上和Scala中的List列表非常像
     * List中常见的map、flatMap、filter、groupBy、foreach等操作在RDD中都有
     * 切记RDD中是不存储数据的！！！
     *
     * 在处理数据的整个流程中，实际上是RDD和RDD之间的转换
     * 不会说调用什么方法之后RDD就变成其它类型了
     * 整个过程都是RDD，所以只需要了解RDD的使用即可掌握SparkCore
     * 统一编程模型
     *
     */
    // 特性1：RDD是由一系列分区Partition组成的，分区数默认等于切片的数量
    // Spark并没有特殊的加载文件的方式，实际上还是用的MR那一套，也会有10%溢出率的考虑
    // 同时切片的数量就会决定RDD的分区数
    val lineRDD: RDD[String] = sc.textFile(inputPath)
    println(s"lineRDD的分区数为：${lineRDD.getNumPartitions}")

    // 将每行数据进行切分，提取每个单词，将每个单词转成一行
    // 特性2：RDD的分区数决定了Task的数量，Task是作用在每个分区上的，每个分区都需要一个Task
    val wordsRDD: RDD[String] = lineRDD.flatMap(line => line.split(","))
    println(s"wordsRDD的分区数为：${wordsRDD.getNumPartitions}")

    // 对每个单词进行分组
    // 特性3：RDD之间是由一系列依赖关系(血缘关系)的，可以分为两种关系：宽依赖、窄依赖
    // 子RDD的分区数默认等于父RDD的分区数
    val grpRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word, 2)
    println(s"grpRDD的分区数为：${grpRDD.getNumPartitions}")

    // 统计每个单词的数量
    val wordCntRDD: RDD[String] = grpRDD.map(kv => s"${kv._1},${kv._2.size}")
    println(s"wordCntRDD的分区数为：${wordCntRDD.getNumPartitions}")

    // 写入前对输出路径进行清理
    // 使用HDFS的API
    val configuration: Configuration = new Configuration()
    val fs: FileSystem = FileSystem.get(configuration)
    if (fs.exists(new Path(outputPath))) {
      fs.delete(new Path(outputPath), true)
    }

    // action行为算子的数量决定了job的数量
    wordCntRDD.saveAsTextFile(outputPath)


  }

}
